import redis
import json
import threading
import time

# 连接Redis
# 增加Redis连接超时时间
r = redis.Redis(
    host='localhost',
    port=6379,
    db=0
)


# 消费者组配置
CONSUMER_GROUP = "plate_recognize_group"
CONSUMER_NAME = "worker_1"  # 可改为动态生成

def init_consumer_group():
    """ 初始化消费者组（只需执行一次） """
    try:
        r.xgroup_create(
            name="task_queue",
            groupname=CONSUMER_GROUP,
            id="$",  # 消费者组初始化时的起始位置标识，接收消费者组创建后新到达的消息
            mkstream=True  # 如果Stream不存在则自动创建
        )
        print(f"Created consumer group: {CONSUMER_GROUP}")
    except redis.exceptions.ResponseError as e:
        if "BUSYGROUP" in str(e):
            print(f"Consumer group {CONSUMER_GROUP} already exists")
        else:
            raise

def consumer_with_ack():


    while True:
        try:
            # 从消费者组读取消息
            messages = r.xreadgroup(
                groupname=CONSUMER_GROUP,
                consumername=CONSUMER_NAME,
                streams={"task_queue": ">"},  # >表示当消费者崩溃重启时，使用>可以继续处理未ACK的消息
                count=1,
                block=5000,
                noack=False  # 必须显式发送ACK
            )

            if not messages:
                continue

            stream, entries = messages[0]
            for msg_id, msg in entries:
                data = None
                try:
                    data = json.loads(msg[b'data'])
                    print(f"[处理中] {msg_id}: {data}")

                    # === 业务处理逻辑 ===
                    # process_data(data)
                    # ====================

                    # 发送ACK确认
                    r.xack("task_queue", CONSUMER_GROUP, msg_id)
                    print(f"[成功] ACK消息 {msg_id}")

                except Exception as e:
                    print(f"[失败] 消息 {msg_id} 处理失败: {str(e)}")
                    # 可添加重试逻辑（示例：最多重试3次）
                    handle_retry(msg_id, data)

        except redis.exceptions.ConnectionError:
            print("Redis连接丢失，5秒后重连...")
            time.sleep(5)

def handle_retry(msg_id, data, max_retries=3):
    """ 消息重试处理 """
    retry_count = r.xinfo_consumers("task_queue", CONSUMER_GROUP)[0].get('pending', 0)

    if retry_count < max_retries:
        print(f"第{retry_count+1}次重试消息 {msg_id}")
    else:
        print(f"消息 {msg_id} 达到最大重试次数，移入死信队列")
        # 将失败消息转移到死信队列
        r.xadd("dead_letter_queue", {"data": json.dumps(data)})
        r.xack("task_queue", CONSUMER_GROUP, msg_id)

# 生产者保持不变
def producer():
    msg_data = {"task_id": 1001, "action": "process_image"}
    msg_id = r.xadd(
        name="task_queue",
        fields={"data": json.dumps(msg_data)},
        maxlen=1000  # 保留最近1000条消息
    )
    print(f"Produced message ID: {msg_id}")

if __name__ == "__main__":
    """ 使用ACK机制的消费者 """
    init_consumer_group()
    threading.Thread(target=consumer_with_ack).start()
    producer()



